package com.mr.reducejoin;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

/* 输入：
 * order.txt：1001	01	1
 * pd.txt：01	小米
 * 输出：
 * 1001	小米	1
 *
 * 1.Mapper阶段无法完成join，只能封装数据，在reduce阶段完成join
 * 2.Bean必须能封装所有的数据
 * 3.没有要求排序，所以将JoinBean放到value-out
 * 4.Reduce只需要输出来自order.txt的数据，需要在Mapper中对数据打标记，标记数据的来源
 * 5.在切片中需要获取当前切片的来源，根据来源执行不同的封装逻辑
 */

public class ReduceJoinMapper extends Mapper<LongWritable, Text, NullWritable, JoinBean> {

	private JoinBean jb = new JoinBean();
	private String source;

	// setup()在map()之前先运行，只运行一次
	@Override
	protected void setup(Mapper<LongWritable, Text, NullWritable, JoinBean>.Context context)
			throws IOException, InterruptedException {

		// 获取内容来自哪个切片 // InputSplit为抽象类，强转为子类
		FileSplit inputSplit = (FileSplit) context.getInputSplit();

		// 获取文件名
		source = inputSplit.getPath().getName();

	}

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, JoinBean>.Context context)
			throws IOException, InterruptedException {



		String[] words = value.toString().split("\t");

		/*
		 * Mapper类run()方法中调用map()方法之前会调用setup()方法。
		 * 可以将下面的内容放到setup()方法中。以免每执行一次map()方法都需要重新执行他们
		 *
		 * // 获取内容来自哪个切片
		 * // InputSplit为抽象类，强转为子类
		 * FileSplit inputSplit = (FileSplit)context.getInputSplit();
		 *
		 * // 获取文件名
		 * source = inputSplit.getPath().getName();
		 */


		// 打标记
		jb.setSource(source);

		// 封装数据
		if (source.equals("order.txt")) {
			jb.setOrderId(words[0]);
			jb.setpId(words[1]);
			jb.setAmount(words[2]);
			// 保证所有的属性不为null，否则会报空指针异常
			jb.setpName("noData");
		} else {
			jb.setpId(words[0]);
			jb.setpName(words[1]);
			jb.setOrderId("noData");
			jb.setAmount("noData");
		}

		context.write(NullWritable.get(), jb);

	}

}
